/****************************************************************************
 *
 * Copyright (C) 2003-2008 Los Alamos National Security, LLC
 *                         Packet Analytics Corporation
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License Version 2 as
 * published by the Free Software Foundation.  You may not use, modify or
 * distribute this program under any other version of the GNU General
 * Public License.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
 *
 ****************************************************************************/
package nfse.query;

import java.io.*;
import java.net.*;
import java.sql.*;
import java.util.*;

import javax.net.SocketFactory;
import javax.net.ssl.*;

import nfse.*;
import nfse.result.*;

public class StaticQueryProcessor {

	public QueryServer queryServer = null;

	private JDBCDatabase statDB = null;

	private NetFSERecord rec = null;

	public StaticQueryProcessor(QueryServer queryServer, int type) {
		this.queryServer = queryServer;
		rec = (NetFSERecord) NetFSE.TYPE_CLASSES.get(new Integer(type));
	}

	public int process(Query query, MetaDatabase mdb) {
		int val = 0;
		Socket resultSocket = null;
		ServerSocket ss = null;
		DataOutputStream resultOut = null;
		SocketFactory sf = SocketFactory.getDefault();
		SocketResult socketResult = null;
		DatabaseResult databaseResult = null;
		MarshalledResult marshalledResult = null;
		SearchEngine se = null;
		try {
			val = query.validate();
			if (val == Query.QUERY_INVALID) {
				System.err.println("Attempted to process an invalid query: '"
						+ query.toXML() + "'");
				return -1;
			}
			if (val == Query.QUERY_EMPTY) {
				System.out.println("Processed an empty query: '"
						+ query.toXML() + "'");
				return 0;
			}
			// System.out.println("Processing query: '" + query.toXML() + "'");
			int numSubs = query.getNumSubqueries();
			if (numSubs > 0) {
				Query[] subs = query.getSubqueries();
				for (int i = 0; i < numSubs; i++) {
					subs[i] = subs[i].detach(this.queryServer.nextQueryID());
					this.process(subs[i], mdb);
				}
				System.gc();
				return 0;
			}

			int queryID = query.getID();
			if (query.getNumDataTypes() == 1) {
				Data datatype = query.getData()[0];
				if (datatype.sensor == -1) {
					Query[] queries = query.decomposeBySensor();
					if (queries.length == 0) {

						return 0;
					}

					ResultMarshall rm = null;
					try {
						rm = Query.marshallResults(queries,
								NetFSE.LOCAL_IP_ADDRESS,
								NetFSE.RESULT_SERVER_PORT, queryID);
					} catch (QueryException qe) {
						qe.printStackTrace();
						return 0;
					}
					this.queryServer.getResultServer().addResultMarshall(rm);
					for (int i = 0; i < queries.length; i++) {
						queries[i] = queries[i].detach(queryServer
								.nextQueryID());
						this.process(queries[i], mdb);
					}
					System.gc();
					return 0;
				}
			}

			Data[] data = query.getData();
			if (data.length != 1) {
				throw (new QueryException(
						"Invalid Data elements specified or failed decomposition."));
			}
			int type = data[0].type;
			int sensor = data[0].sensor;

			DateRange[] dateRanges = query.getDateRanges();
			if ((query.isDistribute())
					&& ((dateRanges.length > 1) || ((dateRanges.length == 1) && (dateRanges[0]
							.isRange())))) {
				Vector<QueryDate> dates = new Vector<QueryDate>();
				Statement s = mdb.createStatement();
				for (int i = 0; i < dateRanges.length; i++) {
					String sql = "";
					if (dateRanges[i].isRange()) {
						sql = "select distinct a.Month, a.Day, a.Year from Static_Indices a, Sensors b where a.Sensor_ID = b.ID and b.ID = "
								+ sensor
								+ " and (a.Min_Time >= "
								+ dateRanges[i].getStartDate().startTS
								/ 1000
								+ " and a.Max_Time <="
								+ dateRanges[i].getEndDate().endTS / 1000 + ")";
					} else {
						sql = "select distinct Month, Day, Year from Static_Indices a, Sensors b where a.Sensor_ID = b.ID and b.ID = "
								+ sensor
								+ " and (a.Min_Time >= "
								+ dateRanges[i].getStartDate().startTS
								/ 1000
								+ " and a.Max_Time <="
								+ dateRanges[i].getStartDate().endTS
								/ 1000
								+ ")";
					}
					// System.out.println(sql);
					/*
					 * if (dateRanges[i].isRange()) { sql = "select distinct
					 * a.Month, a.Day, a.Year from Static_Indices a, Sensors b
					 * where a.Sensor_ID = b.ID and b.ID = " + sensor + " and
					 * a.Index_Date between '"; sql +=
					 * dateRanges[i].getStartDate().toMySQLString() + "' and '"
					 * + dateRanges[i].getEndDate().toMySQLString() + "' order
					 * by a.Index_Date"; } else { sql = "select distinct Month,
					 * Day, Year from Static_Indices a, Sensors b where
					 * a.Sensor_ID = b.ID and b.ID = " + sensor + " and
					 * a.Index_Date='"; sql +=
					 * dateRanges[i].getStartDate().toMySQLString() + "' order
					 * by a.Index_Date"; }
					 */
					ResultSet rs = s.executeQuery(sql);
					while (rs.next()) {
						int m = rs.getInt("Month");
						int d = rs.getInt("Day");
						int y = rs.getInt("Year");
						dates.addElement(new nfse.query.QueryDate(m, d, y));
					}
					rs.close();
					sql = "";
					/*
					 * if (dateRanges[i].isRange()) { sql = "select distinct
					 * a.Month, a.Day, a.Year from Local_MySQL_Indices a,
					 * Sensors b where a.Sensor_ID = b.ID and b.ID =
					 * " + sensor + " and a.Index_Date between '"; sql +=
					 * dateRanges[i].getStartDate().toMySQLString() + "' and '"
					 * + dateRanges[i].getEndDate().toMySQLString() + "' "; }
					 * else { sql = "select distinct Month, Day, Year from
					 * Local_MySQL_Indices a, Sensors b where a.Sensor_ID = b.ID
					 * and b.ID = " + sensor + " and a.Index_Date='"; sql +=
					 * dateRanges[i].getStartDate().toMySQLString() + "' "; } rs
					 * = s.executeQuery(sql); while (rs.next()) { int m =
					 * rs.getInt("Month"); int d = rs.getInt("Day"); int y =
					 * rs.getInt("Year"); NetFSE.query.Date date = new
					 * NetFSE.query.Date(m, d, y); boolean found = false; for
					 * (int x = 0; x < dates.size(); x++) { NetFSE.query.Date
					 * date2 = (NetFSE.query.Date) dates .elementAt(x); if
					 * (date2.compareTo(date) == 0) { found = true; break; } }
					 * if (!found) dates.addElement(date); } rs.close();
					 */
				}
				s.close();

				nfse.query.QueryDate[] sortedDates = new nfse.query.QueryDate[dates
						.size()];
				for (int i = 0; i < dates.size(); i++) {
					sortedDates[i] = (nfse.query.QueryDate) dates.elementAt(i);
				}
				Arrays.sort(sortedDates);
				NodeStatus[] queryNodes = NetFSE.getQueryNodeStatus();
				int datesPerNode = 0;
				if (queryNodes.length == 0)
					datesPerNode = sortedDates.length;
				else
					datesPerNode = sortedDates.length / queryNodes.length;
				if ((queryNodes.length != 0)
						&& (sortedDates.length % queryNodes.length != 0))
					datesPerNode++;

				try {
					socketResult = (SocketResult) query.getResult();
				} catch (ClassCastException cce) {
					try {
						databaseResult = (DatabaseResult) query.getResult();
					} catch (ClassCastException cce2) {
						marshalledResult = (MarshalledResult) query.getResult();
					}
				}

				if (datesPerNode == 0) {

					if (socketResult != null) {
						resultSocket = sf.createSocket(socketResult.host,
								socketResult.port);
						resultOut = new DataOutputStream(resultSocket
								.getOutputStream());
					} else if (marshalledResult != null) {
						resultSocket = sf.createSocket(marshalledResult.host,
								marshalledResult.port);
						resultOut = new DataOutputStream(resultSocket
								.getOutputStream());
						resultOut.writeInt(marshalledResult.queryID);
						resultOut.writeInt(NetFSE.RESULT_SERVER_FLAG_DONE);
						resultOut.writeInt(-100);
						resultOut.writeInt(0);
						DataInputStream resultIn = new DataInputStream(
								resultSocket.getInputStream());
						resultIn.readInt();
						resultIn.close();
					}
					if (resultOut != null) {
						resultOut.close();
						resultSocket.close();
					}
					return 0;
				}

				int numNodes = sortedDates.length / datesPerNode;
				if (sortedDates.length % datesPerNode != 0)
					numNodes++;
				if (datesPerNode == 1)
					numNodes = sortedDates.length;

				Query[] queries = new Query[numNodes];
				for (int i = 0; i < numNodes; i++) {
					Query q = ((Query) query.clone()).detach(queryServer
							.nextQueryID());
					q.setDate(null);
					q.setDistribute(false);
					int min = i * datesPerNode;
					int max = i * datesPerNode + datesPerNode;
					Vector<DateRange> temp = new Vector<DateRange>();
					for (int j = min; j < max; j++) {
						if (j < sortedDates.length) {
							temp.addElement(new DateRange(sortedDates[j]));
						}
					}
					DateRange[] dates2 = DateRange.consolidate(temp);

					for (int j = 0; j < dates2.length; j++) {
						q.addDateRange(dates2[j]);
					}
					queries[i] = q;
				}
				if (queries.length > 1) {
					ResultMarshall rm = Query.marshallResults(queries,
							NetFSE.LOCAL_IP_ADDRESS, NetFSE.RESULT_SERVER_PORT,
							query.getID());
					this.queryServer.getResultServer().addResultMarshall(rm);
					Vector<String> tempNames = new Vector<String>();
					for (int i = 0; i < queryNodes.length; i++) {
						if (queryNodes[i].getName()
								.compareTo(NetFSE.LOCAL_HOST) != 0)
							tempNames.addElement(queryNodes[i].getName());
						else
							tempNames.addElement("localhost");
					}
					String[] nodes = new String[tempNames.size()];
					for (int i = 0; i < nodes.length; i++)
						nodes[i] = (String) tempNames.elementAt(i);
					if (nodes.length == 0) {
						nodes = new String[1];
						nodes[0] = "localhost";
					}
					for (int i = 0; i < queries.length; i++) {
						queries[i] = queries[i].detach(queryServer
								.nextQueryID());
						if (nodes[i % nodes.length].compareTo("localhost") == 0) {
							this.process(queries[i], mdb);
						} else {
							SocketFactory sf2 = null;
							if (NetFSE.USE_SSL_QUERY)
								sf2 = SSLSocketFactory.getDefault();
							else
								sf2 = SocketFactory.getDefault();
							Socket sock = null;
							try {
								sock = sf2.createSocket(
										nodes[i % nodes.length],
										NetFSE.QUERY_SERVER_PORT);
							} catch (IOException ioe) {
								System.err.println("Node '"
										+ nodes[i % nodes.length]
										+ "' not responding.");
							}
							if (sock != null) {
								PrintWriter socketOut = new PrintWriter(sock
										.getOutputStream());
								DataInputStream socketIn = new DataInputStream(
										sock.getInputStream());
								socketOut.println(queries[i].toXML());
								System.out.println("Sending query to "
										+ nodes[i % nodes.length] + ": "
										+ queries[i].toXML());
								socketOut.println("<--DiSARM: end query-->");
								socketOut.flush();
								socketIn.readInt();
								socketOut.close();
								socketIn.close();
								sock.close();

							} else {
								this.process(queries[i], mdb);
							}

						}
					}
					return 0;
				} else
					query = queries[0];
			}

			// query.generateRecordFilter();
			try {
				socketResult = (SocketResult) query.getResult();
			} catch (ClassCastException cce) {
				try {
					databaseResult = (DatabaseResult) query.getResult();
				} catch (ClassCastException cce2) {
					marshalledResult = (MarshalledResult) query.getResult();
				}
			}
			// RecordShaper recordShaper = query.getRecordShaper();
			// RecordFilter recordFilter = query.getRecordFilter();
			if (socketResult != null) {
				// System.out.println("SOCKET RESULT " + socketResult.host + ":"
				// + socketResult.port);
				resultSocket = sf.createSocket(socketResult.host,
						socketResult.port);
				resultSocket.setSendBufferSize(NetFSE.SOCKET_SEND_SIZE);
				resultOut = new DataOutputStream(resultSocket.getOutputStream());
			} else {
				// System.out.println("MARSHALLED RESULT " +
				// marshalledResult.host + ":" + marshalledResult.port);
				resultSocket = sf.createSocket(marshalledResult.host,
						marshalledResult.port);
				resultSocket.setSendBufferSize(NetFSE.SOCKET_SEND_SIZE);
				resultOut = new DataOutputStream(resultSocket.getOutputStream());
				resultOut.writeInt(marshalledResult.queryID);
				resultOut.writeInt(NetFSE.RESULT_SERVER_FLAG_DONE);
			}

			boolean done = false;
			DateRange[] dates = query.getDateRanges();
			TimeRange[] times = query.getTimeRanges();
			TimestampRange[] timestampRanges = query.getTimestampRanges();
			if (times.length > 1)
				throw (new QueryException(
						"Support for multiple time ranges currently not implemented."));
			if (dates.length > 0) {
				Vector<DateRange> tempDateRanges = new Vector<DateRange>();
				for (int i = 0; i < dates.length; i++) {
					nfse.query.QueryDate[] tempDates = dates[i].getDates();
					for (int j = 0; j < tempDates.length; j++)
						tempDateRanges.addElement(new DateRange(tempDates[j]));
				}
				dates = new DateRange[tempDateRanges.size()];
				for (int i = 0; i < dates.length; i++)
					dates[i] = (DateRange) tempDateRanges.elementAt(i);
			}

			Criteria[] criteria = query.getCriteria();
			int[] predicates = new int[criteria.length + 1];
			String[] values1 = new String[criteria.length + 1];
			String[] values2 = new String[criteria.length + 1];
			GlobalField[] fields = new GlobalField[criteria.length + 1];
			fields[fields.length - 1] = NetFSE
					.getGlobalField(NetFSE.FIELD_START_TS);
			predicates[predicates.length - 1] = NetFSE.PREDICATE_RANGE;
			for (int i = 0; i < criteria.length; i++) {
				GlobalField f = NetFSE.getGlobalField(criteria[i].field);

				if (f == null) {
					System.err.println("StaticQueryProcessor: Field was null.");
					resultOut.close();
					resultSocket.close();
					return -1;
				} else
					fields[i] = f;
				predicates[i] = criteria[i].predicate;
				if (criteria[i].predicate == NetFSE.PREDICATE_RANGE) {
					values1[i] = NetFSEUtil.convertToHex(criteria[i].field,
							criteria[i].value1);
					values2[i] = NetFSEUtil.convertToHex(criteria[i].field,
							criteria[i].value2);
				} else {
					values1[i] = NetFSEUtil.convertToHex(criteria[i].field,
							criteria[i].value1);
				}
			}
			byte[] bytes = null;
			int count = 0;

			Class.forName("org.postgresql.Driver");
			String url = "jdbc:postgresql://localhost/"
					+ NetFSE.getTypeName(type).toLowerCase();
			Properties props = new Properties();
			props.setProperty("user", NetFSE.DYNAMIC_DB_USER);
			props.setProperty("password", NetFSE.DYNAMIC_DB_PASSWORD);
			props.setProperty("client_encoding", "UTF8");

			Connection tempConnection = null;
			Statement tempStatement = null;
			try {
				tempConnection = DriverManager.getConnection(url, props);
			} catch (Exception e) {
				url = "jdbc:postgresql://localhost/" + NetFSE.METADATA_DATABASE;
				props.setProperty("user", NetFSE.DYNAMIC_DB_USER);
				props.setProperty("password", NetFSE.DYNAMIC_DB_PASSWORD);
				tempConnection = DriverManager.getConnection(url, props);
				tempStatement = tempConnection.createStatement();
				tempStatement.execute("create database "
						+ NetFSE.getTypeName(type).toLowerCase());

				tempStatement.close();
				tempConnection.close();

				url = "jdbc:postgresql://localhost/"
						+ NetFSE.getTypeName(type).toLowerCase();
				props.setProperty("user", NetFSE.DYNAMIC_DB_USER);
				props.setProperty("password", NetFSE.DYNAMIC_DB_PASSWORD);

			}
			tempStatement = tempConnection.createStatement();
			for (int y = 0; y < timestampRanges.length; y++) {
				// System.out.println("TIMESTAMP SEARCH");
				// System.out.println(timestampRanges[y].toXML());
				Run[] runs = mdb.getRuns(timestampRanges[y].startTS,
						timestampRanges[y].endTS, sensor);
				for (int x = 0; x < runs.length; x++) {
					// System.out.println("\t" + runs[x].toString());
					// System.out.println("Min_Time: " + runs[x].minTime);
					// System.out.println("Start_Time: " +
					// timestampRanges[y].startTS.getTime() / 1000);
					// System.out.println("Max_Time: " + runs[x].maxTime);
					// System.out.println("Stop_Time: " +
					// timestampRanges[y].endTS.getTime() / 1000);

					// boolean srcIPExists = true;
					// if (srcIP.length() > 0)
					// srcIPExists = valueExistsInRunSrcIP(type, runs[x],
					// srcIP);
					// boolean dstIPExists = true;
					// if (dstIP.length() > 0)
					// dstIPExists = valueExistsInRunDstIP(type, runs[x],
					// dstIP);

					// System.out.println("Found match for IP in stats database:
					// " + srcIP);

					// if ((srcIPExists) && (dstIPExists)) {
					// System.out.println("Searching " +
					// runs[x].toString());
					boolean droppedTimeSearch = false;
					if ((runs[x].getMinTime() >= timestampRanges[y].startTS
							.getTime() / 1000)
							&& (runs[x].getMaxTime() <= timestampRanges[y].endTS
									.getTime() / 1000)) {

						droppedTimeSearch = true;
					}
					values1[values1.length - 1] = null;
					values2[values2.length - 1] = null;

					se = new SearchEngine(mdb, tempConnection, rec);
					se.performSearch(runs[x].getMonth(), runs[x].getDay(),
							runs[x].getYear(), runs[x].getRun(), type, sensor,
							fields, values1, values2, predicates);

					String line = null;
					NetFSERecord record = NetFSE.getTypeClass(type);
					line = se.getNextRecord(tempStatement);
					// System.out.println(line);
					long trStartTS = timestampRanges[y].startTS.getTime();
					long trEndTS = timestampRanges[y].endTS.getTime();
					while (line != null) {
						if (droppedTimeSearch) {
							count++;
							bytes = line.getBytes();
							try {
								resultOut.writeInt(type);
								resultOut.writeInt(sensor);
								resultOut.writeInt(bytes.length);
								resultOut.write(bytes);
							} catch (IOException ioe) {
								line = null;
								done = true;
							}
						} else {
							NetFSERecord rec = null;
							try {
								rec = record.generate(line, sensor);
							} catch (Exception e) {
								System.out.println(line);
								e.printStackTrace();
							}
							if (rec != null) {
								long startts = rec.getStartTS().getTime();
								if (startts >= trStartTS && startts <= trEndTS) {
									count++;
									bytes = line.getBytes();
									try {
										resultOut.writeInt(type);
										resultOut.writeInt(sensor);
										resultOut.writeInt(bytes.length);
										resultOut.write(bytes);
									} catch (IOException ioe) {
										line = null;
										done = true;
									}
								}
							}
						}

						if (line != null)
							line = se.getNextRecord(tempStatement);
					}

					se.close();
				}
			}

			for (int y = 0; y < dates.length; y++) {
				if (!done) {
					// System.out.println("DATE SEARCH");
					Run[] runs = mdb.getRuns(dates[y].getStartDate(), dates[y]
							.getEndDate(), sensor);
					for (int x = 0; x < runs.length; x++)
						System.out.println(runs[x]);
					for (int x = 0; x < runs.length; x++) {
						if (done)
							break;

						if (data.length > 1) {
							System.err
									.println("Number of data types greater than 1. Not good!");
						} else if (data.length == 0) {
							System.err
									.println("Number of data types is 0. Really not good!");
							return -1;
						}

						se = new SearchEngine(mdb, tempConnection, rec);
						se.performSearch(runs[x].getMonth(), runs[x].getDay(),
								runs[x].getYear(), runs[x].getRun(), type,
								sensor, fields, values1, values2, predicates);

						String line = null;
						if (databaseResult == null) {
							line = se.getNextRecord(tempStatement);

							while (line != null) {
								count++;
								bytes = line.getBytes();
								try {
									resultOut.writeInt(type);
									resultOut.writeInt(sensor);
									resultOut.writeInt(bytes.length);
									resultOut.write(bytes);
								} catch (IOException ioe) {
									line = null;
									done = true;
								}

								if (line != null)
									line = se.getNextRecord(tempStatement);
							}

						} else { // Database result

							// Deprecated

						}
						se.close();

					}
				}
			}
			tempStatement.close();
			tempConnection.close();

			if (resultOut != null) {
				if (marshalledResult != null) {
					resultOut.writeInt(-100);
					resultOut.writeInt(count);
					DataInputStream resultIn = new DataInputStream(resultSocket
							.getInputStream());
					resultIn.readInt();
					resultIn.close();
					System.out
							.println("StaticQueryProcessor2: finished query #"
									+ marshalledResult.queryID + ", records="
									+ count);
				}
				resultOut.close();
				resultSocket.close();
			}
			if (statDB != null)
				statDB.closeConnection();
		} catch (Exception E) {
			E.printStackTrace();
			System.out.println("Client closed connection; cleaning up.");
			try {
				if (se != null) {
					se.close();
				}
				if (socketResult != null) {
					resultSocket = sf.createSocket(socketResult.host,
							socketResult.port);
					resultOut = new DataOutputStream(resultSocket
							.getOutputStream());
				} else if (marshalledResult != null) {
					resultSocket = sf.createSocket(marshalledResult.host,
							marshalledResult.port);
					resultOut = new DataOutputStream(resultSocket
							.getOutputStream());
					resultOut.writeInt(marshalledResult.queryID);
					resultOut.writeInt(NetFSE.RESULT_SERVER_FLAG_DONE);
					resultOut.writeInt(-100);
					resultOut.writeInt(0);
					DataInputStream resultIn = new DataInputStream(resultSocket
							.getInputStream());
					resultIn.readInt();
					resultIn.close();
				}
				if (resultOut != null) {
					resultOut.close();
					resultSocket.close();
				}

			} catch (Exception e2) {
				e2.printStackTrace();
			}
		}

		try {
			if (ss != null)
				ss.close();
			if (resultSocket != null)
				resultSocket.close();
		} catch (IOException ioe) {
			ioe.printStackTrace();
		}
		return 0;
	}

}